#include "luredis.hpp"
#include "l_utility.hpp"
#include <stdlib.h>

#define __REDIS_MULTI_CMD "MULTI"
#define __REDIS_EXEC_CMD "EXEC"
#define __REDIS_CONTEXT "redis_context"
#define __LUREDIS "luredis"
#define __REDIS_REPLY_TYPE "reply_type"
#define __LUREDIS_ARG_ERROR 0

namespace http {
    namespace server3 {

    /*
      luredis support 1 internal command buffer
      To interact with the internal command buffer use:
            - send_cmd_buffer (redisContext c, format/nil if not used)
            - append_cmd_buf
            - get_cmd_buf
      To send command directly to redis server without using command buffer use:
            - send_cmd (redisContext c, format/nil if not used, command/data)
      */

static void moveReplyObjectToLuaStack(lua_State *L, redisReply *reply);
static int luredis_exe_cmd(lua_State* L, redisContextPtr *c, const char* format, const char*cmd);

static int luredis_connect(lua_State* L){
    // timeout is set to 1.5 seconds
    struct timeval timeout;

    // the default address is "127.0.0.1"
    const char* host = luaL_optstring(L, 1, "127.0.0.1");
    // the default port is <int>6379
    int port = luaL_optint(L, 2, 6379);
    int timeout_secs = luaL_optint(L, 3, 1);
    timeout.tv_sec = timeout_secs;
    timeout.tv_usec = 0;

    // create a new variable of type redisContextPtr
    // Then, it  is pushed to the top of LUA stack
    redisContextPtr *c= (redisContextPtr*) lua_newuserdata(L, sizeof(redisContextPtr));
    *c = redisConnectWithTimeout(host, port, timeout);
    if ((*c)->err)
    {
#ifdef DEBUG_LUREDIS
        printf("[CONNECTION ERROR] At <%s:%d> :%s\n",host,port,(*c)->errstr);
        printf("Exiting ... \n");
#endif
        /* Free Redis Context in case it is unable to connect to redis server */
        redisFree(*c);
        return 0;
    }
    // the metatable describe behaviors of the userdata
    // and must be defined at the time we open luredis library
    luaL_getmetatable(L,__REDIS_CONTEXT);
    // the metatable is now at the top of the stack (the highest index),
    // which is at (L,-1). Also, the userdata is now at (L,-2)
    // Set the metatable (at the top of the stack) associated with
    // the userdata (at (L,-2))
    lua_setmetatable(L,-2);
    return 1;
}

static redisContextPtr *checkRedisContext (lua_State *L, int index)
{
  redisContextPtr *c = NULL;
  // query to know whether the data at (L,index) is of type LUA_TUSERDATA
  if(lua_isuserdata(L,index))
     c = (redisContextPtr *)luaL_checkudata(L, index, __REDIS_CONTEXT);
#ifdef DEBUG_LUREDIS
  else
      printf("Checked data is not of type redisContext\n");
#endif
  return c;
}

static int luredis_command(lua_State* L){
    int argc = lua_gettop(L);
    if(argc < 2)  {
        lua_pushnumber(L,__LUREDIS_ARG_ERROR);
        return 1;
    }
    /* redis_command(redisContext c, command, format  */
    redisContextPtr *c;
    c = checkRedisContext(L, 1);
    if(c == NULL) return 0;
    /* (1) redisContext, (2) Format */
    if(lua_isnoneornil(L,3))
        return luredis_exe_cmd(L,c,luaL_checkstring(L,2),NULL);
    else
        return luredis_exe_cmd(L,c,luaL_checkstring(L,2),luaL_checkstring(L,3));
}

static int luredis_send_buffer(lua_State* L){
    // TODO:
    // we don't use pipelining as the context may die,
    //  which cause the lost of the piple
    // save command buffer
    int argc = lua_gettop(L);
    if(argc < 1) {
        lua_pushnumber(L,__LUREDIS_ARG_ERROR);
        return 1;
    }
    redisContextPtr *c = NULL;    
    /* Send command */
    c = checkRedisContext(L, 1);
    if(c == NULL) return 0;

    if(lua_isnoneornil(L,2))
        return luredis_exe_cmd(L,c,l_get_wq_buf(L),NULL);
    else
        return luredis_exe_cmd(L,c,luaL_checkstring(L,2),l_get_wq_buf(L));
}

static int luredis_exe_cmd(lua_State* L, redisContextPtr *c, const char* format, const char*cmd) {
    redisReply *reply;
    if(cmd == NULL)
        reply = (redisReply*)redisCommand(*c,format);
    else
        reply = (redisReply*)redisCommand(*c,format,cmd);
    if(reply == NULL)
    {
        if(*c!=NULL) {
            redisFree(*c);
            *c = NULL;
        }
#ifdef DEBUG_LUREDIS
        printf("[CONNECTION ERROR] Redis server do not reply\n");
#endif
        return 0;
    }
    /* Push returned data to LVM */
    lua_pushnumber(L,reply->type);
#ifdef DEBUG_LUREDIS
    printf("Status = %d\n", reply->type);
#endif
    /* NOTICE: moveReplyObjectToLuaStack pushes 1 value onto Lua Stack */
    moveReplyObjectToLuaStack(L,reply);
    return 2;

}

#ifdef ___________________________
static redisReply *helper_pipelining_command(redisContextPtr *c, list_rd* pipe)
{
    // travel the list and send data
    listIter *iter;
    listNode *fmatNode, *cmdNode;
    iter = listGetIterator(pipe, AL_START_HEAD);
    // if we use pipeline, the returned reply object is unpredictable
    while(1){
        fmatNode = listNext(iter);
        if(fmatNode == NULL) break;
        cmdNode = listNext(iter);
        redisReply *reply = (redisReply*)redisCommand(*c,(sds)(fmatNode->value),(sds)(cmdNode->value));
        freeReplyObject(reply);
        listDelNode(pipe,fmatNode);
        listDelNode(pipe,cmdNode);
#ifdef DEBUG_LUREDIS
        printf("Deleted node\n. List length = %d\n",pipe->len);
#endif
    }
    listReleaseIterator(iter);
    //redisCommand(*c,__REDIS_EXEC_CMD);
    return NULL;//(redisReply*)redisCommand(*c,__REDIS_EXEC_CMD);
}
#endif

static int luredis_clear_buffer(lua_State* L) {
    sdsmakeempty(l_get_wq_buf(L));
    return 0;
}
static int luredis_reset_buffer(lua_State* L) {
    sdsmakeempty(l_get_wq_buf(L));
    sdsupdatelen(l_get_wq_buf(L));
    return 0;
}
static int luredis_append_buffer(lua_State* L){
    // An intelligent function
    int argc = lua_gettop(L);    
    sds cmd_buf = l_get_wq_buf(L);
    if(argc>0){
        for(int i = 1; i < argc+1; i++){
            if(lua_isstring(L,i)) {
                cmd_buf = sdscat(cmd_buf, luaL_checkstring(L, i));
                if(cmd_buf == NULL) {
                    cmd_buf = sdsallocate(1024);
                    break;
                }
            }
        }
        l_set_wq_buf(L,cmd_buf);
    }
    return 0;
}

static int luredis_concat_cmd_buf(lua_State* L){
    // An intelligent function
    int argc = lua_gettop(L);
    sds cmd_buf = l_get_wq_buf(L);
    const char* seperator;
    if(argc>0){
        if(lua_isstring(L,1)) {
            seperator= luaL_checkstring(L, 1);
        }
        else return 0;
        for(int i = 2; i < argc; i++){
            if(lua_isstring(L,i)) {
                if((cmd_buf = sdscat(cmd_buf, luaL_checkstring(L, i))) == NULL)
                    cmd_buf = sdsallocate(1024);
                if((cmd_buf = sdscat(cmd_buf, seperator)) == NULL)
                    cmd_buf = sdsallocate(1024);
                }
        }
        if(lua_isstring(L,argc)) {
            if((cmd_buf = sdscat(cmd_buf, luaL_checkstring(L, argc))) == NULL)
                cmd_buf = sdsallocate(1024);
        }
        l_set_wq_buf(L,cmd_buf);
    }
    return 0;
}

static int luredis_get_buffer(lua_State* L){
    lua_pushstring(L,l_get_wq_buf(L));
    return 1;
}

static int luredis_gc (lua_State *L)
{
    // when the garbage collector run
    // its argument, the will-be-collected one, is at the top of LUA stack
  redisContextPtr *c = checkRedisContext(L, 1);
#ifdef DEBUG_LUREDIS
  printf("[GARBAGE COLLECTOR] Cleaned the redisContext at %p\n", *c);
#endif

  if(c != NULL)   redisFree(*c);
  return 0;
}


static const luaL_reg luredis_methods[] = {
  {"new_connector",       luredis_connect},
  {"send_cmd", luredis_command},
  {"append_buffer", luredis_append_buffer},
  {"concat_buffer", luredis_concat_cmd_buf},
  {"get_buffer", luredis_get_buffer},
  {"send_buffer", luredis_send_buffer},
  {"free_connector", luredis_gc},
  {"clear_buffer",luredis_clear_buffer},
  {"reset_buffer",luredis_reset_buffer},
  {NULL, NULL}
};

static Util_Table_SI reply_typeArr[] = {
    {"argerr",__LUREDIS_ARG_ERROR}, /* Argument error */
    {"string",REDIS_REPLY_STRING},
    {"array",REDIS_REPLY_ARRAY},
    {"integer", REDIS_REPLY_INTEGER},
    {"nil",REDIS_REPLY_NIL},
    {"status",REDIS_REPLY_STATUS},
    {"error",REDIS_REPLY_ERROR},
    {NULL,0}
};

int register_luredis (lua_State *L)
{
    luaL_newmetatable(L,__REDIS_CONTEXT); // for userdata
    luaL_register(L, __LUREDIS, luredis_methods);  ;
    l_register_table_SI(L, -1,reply_typeArr,__REDIS_REPLY_TYPE);
    return 1;
}


static void moveReplyObjectToLuaStack(lua_State *L, redisReply *reply) {
    redisReply *r = reply;
    size_t j;
    switch(r->type) {
    case REDIS_REPLY_INTEGER:
        lua_pushnumber(L,reply->integer);
        break; /* Nothing to free */
    case REDIS_REPLY_ARRAY:
        if (r->element != NULL) {
            /* Table; check lua stack */
            // luaL_checkstack(L,sz,msg);
            lua_newtable(L);
            for (j = 0; j < r->elements; j++)
                if (r->element[j] != NULL) {
                    lua_pushnumber(L,j+1);
                    //lua_pushstring(L,r->element[j]->str);
                    moveReplyObjectToLuaStack(L, r->element[j]);
                    lua_rawset(L,-3);
                }
            free(r->element);
        }
        break;
    case REDIS_REPLY_ERROR:
    case REDIS_REPLY_STATUS:
    case REDIS_REPLY_STRING:
        if (r->str != NULL) {
            lua_pushstring(L,r->str);
            free(r->str);
        }
        else lua_pushnil(L);
        break;        
    //case REDIS_REPLY_NIL:
    default:
        lua_pushnil(L);
        break;
    }
    free(r);
}
    }
}
